#include "dbworker.h"
#include "list.h"
#include "th.h"
#include "task.h"
#include "worker_mgr.h"
#include "fix_sche.h"
#include "worker.h"


#include <malloc.h>
#include <assert.h>

C_CODE_BEGIN
enum dbworker_cmd {
	DB_SLEEP, 
	DB_EXIT,
	DB_EXECUTE,
	DB_FINALIZE
};

struct dbworker {
	/*
	 * Thread handler for this dbworker
	 */
	thread_handler th;
	/*
	 * spinlock , this is typically for the pointer swap
	 */
	spinlock plk;
	/*
	 * this lock is used to indicating the sleep status of
	 * worker !
	 */
	fastlock slplk;
	event slpcond;
	
	volatile enum dbworker_cmd cmd;
	/*
	 * Indicating the front list queue for this worker as well as the end queue 
	 */
	struct list* volatile front,* volatile back;
	
	bool detach;
	
};

#define DBWORKER(w) \
	((struct dbworker*)((char*)(w) + sizeof(struct worker)))
	
#define WORKER(dw) \
	((struct worker*)((char*)(dw) - sizeof(struct worker)))


static void _swap_back_front( struct dbworker* wo ) {
	struct list* tmp;
	// just swap the back and front worker queue
	// get the spinlock since this operation is very quick
	thread_spinlock_require( &(wo->plk) , false );
	// swap the pointer
	tmp = wo->front;
	wo->front = wo->back;
	wo->back = tmp;
	// release the lock
	thread_spinlock_release( &(wo->plk) );	
}


static void _finalize ( struct dbworker* wo ) {
	int i = 0 ; 
	for( ; i < 2 ; ++i ) {
		while( list_get_usesize( wo->back) !=0 )  {
		 struct task* eve = ( struct task* ) list_front( wo->back );
		 void* rval;
		 list_pop_front( wo->back );
		 rval = eve->work( eve->par );
		 if ( eve->clean )
			eve->clean( eve->par , rval);
		}
		_swap_back_front( wo );
	}
}

static bool _goto_sleep( struct dbworker* wo) {
	
	// before go to sleep , the thread should check whether a quit signal is generated or not
	 thread_fastlock_require( &(wo->slplk) , NULL );
	 if( wo->cmd == DB_EXIT ) {
		 /* exit the thread right-now */
		 thread_fastlock_release( &(wo->slplk ) );
		 return false;
	 } else if( wo->cmd == DB_FINALIZE ) {
		 _finalize(wo);
		 thread_fastlock_release( &(wo->slplk ) );
		 return false;
	 }
	 // now I am sure the condition will not change
	 // I can goto sleep ,even if a immdiet message
	 // will show up to make me work again .
	 wo->cmd = DB_SLEEP;
	 thread_event_wait( &(wo->slpcond),  &(wo->slplk) , NULL );
	 if(  wo->cmd == DB_EXIT ) {
		 thread_fastlock_release( &(wo->slplk) );
		 return false;
	 } else if( wo->cmd == DB_FINALIZE ) {
		 _finalize(wo);
		 thread_fastlock_release( &(wo->slplk ) );
		 return false;
	 } else {
		  wo->cmd = DB_EXECUTE;
		 thread_fastlock_release( &(wo->slplk) );
		 return true;
	 }
}

static void _dbworker_free( struct worker* w) {
	struct dbworker* dw = DBWORKER(w);
	thread_fastlock_destroy( &(dw->slplk) );
	thread_spinlock_destroy( &(dw->plk) );
	thread_event_destroy( &(dw->slpcond) );
	list_destroy( dw->front );
	list_destroy( dw->back );
	if( dw->detach )
		free(w);
}

static void* _dbworker_main( void* db ) {
	struct dbworker * wo = (struct dbworker*)(db);
	void * rval;
	 while ( 1 ) {
__WORK:
		 //1. When we reach here , means there must be some work we need to do 
		 while( list_get_usesize( wo->back) !=0 )  {
			 struct task* eve = ( struct task* ) list_front( wo->back );
			 rval = eve->work( eve->par );
			 if ( eve->clean )
				eve->clean( eve->par , rval);
			 list_pop_front( wo->back );
		 }
		 //2. No work in the current loop . OK let's swap the pointer here
		 _swap_back_front( wo );
		 if( list_get_usesize(wo->back) == 0 ) {
			 // ok sleep 
			 goto __SLEEP;
		 } else {
			 // ok work now.
			 goto __WORK;
		 }
__SLEEP:
		if( !_goto_sleep(wo) ) 
			goto __FREE;
	}

__FREE:
	_dbworker_free(WORKER(wo));
	return NULL;
}



static bool _dbworker_join( struct worker* w , const struct th_join_sec* tm ) {
	bool ret;
	struct dbworker* dw = DBWORKER(w);
#ifndef NDEBUG
	/*
	 * Try to find out that if this worker can be joined .
	 * Just to check the cmd tag of the worker is DB_EXIT , DB_FINALIZE 
	 */
	 thread_fastlock_require( &(dw->slplk) , NULL );
	 assert( dw->cmd == DB_EXIT || dw->cmd == DB_FINALIZE );
	 thread_fastlock_release( &(dw->slplk) );
#endif
	assert( dw->detach == false );
	ret = thread_join( dw->th , NULL , tm );
	if( ret  )
		free(w);
	return ret ;
	
}

static void* _dbworker_dispatch( struct worker* w , const struct task* t ) {
	struct dbworker* dw = DBWORKER(w);
	/* if the worker is in finalized stage , the user CAN NOT BE ALLOWED to push the task */
	assert( dw->cmd != DB_FINALIZE );
	thread_spinlock_require( &(dw->plk) , false );
	list_push_back( dw->front , (void*)t );
	thread_spinlock_release( &(dw->plk) );
	/*
	 * this is a big bottleneck for the multi-thread operation ,
	 * since every task you push into the queue will send a notification
	 * to notify the worker to wake-up , it is extremely inefficient.
	 */
	thread_fastlock_require( &(dw->slplk) , NULL );
	if( dw->cmd == DB_SLEEP ) {
		thread_event_signal( &(dw->slpcond) );
	}
	thread_fastlock_release( &(dw->slplk) );
	return NULL;
}

static size_t _dbworker_get_workload( struct worker* w ) {
	return list_get_usesize( DBWORKER(w)->front ) + list_get_usesize( DBWORKER(w)->back ) ;
}

static void _dbworker_signal_quit( struct worker* w , bool right_now ) {
	struct dbworker* dw = DBWORKER(w);
	thread_fastlock_require(&(dw->slplk) , NULL );
	/* this message is a manually pushed message */
	dw->cmd = right_now ? DB_EXIT :DB_FINALIZE;
	thread_event_signal( &(dw->slpcond) );
	thread_fastlock_release( &(dw->slplk) );
}

static bool _dbworker_ctl( struct worker* w , const char* cmd , ... ) {
	return false;
}

static struct worker * _dbworker_create( const char* name , bool detach ) {
	struct worker* w = malloc( sizeof( struct dbworker ) + sizeof(struct worker) );
	struct dbworker* dw = DBWORKER(w);
	
	dw->front = list_create( sizeof(struct task) , DBWORKER_CACHE_SIZE );
	dw->back = list_create( sizeof(struct task) , DBWORKER_CACHE_SIZE );
	/*
	 * Initialize the worker lock here
	 */
	 thread_spinlock_create( & (dw->plk) );
	 thread_fastlock_create( & (dw->slplk) );
	 thread_event_create( &(dw->slpcond) );
	 dw->detach = detach;
	 dw->cmd = DB_EXECUTE;
	 /*
	  * create the target thread
	  */
	  if ( thread_create( &(dw->th) , name , _dbworker_main ,  dw , detach ) != true ) {
		  dw->detach = true;
		  _dbworker_free(w);
		  return NULL;
	  }
	  /*
	   * make the operator
	   */
	  w->join = _dbworker_join;
	  w->dispatch = _dbworker_dispatch;
	  w->signal_quit = _dbworker_signal_quit;
	  w->get_workload = _dbworker_get_workload;
	  return w;
} 


void _dbworker_register( struct worker_mgr* mgr ) {
	worker_mgr_register(mgr,DBWORKER_NAME,_dbworker_create);
}

#undef DBWORKER
#undef WORKER

C_CODE_END
